package com.bigdata;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @className: LogTypeInterceptor
 * @description: TODO 通过flume拦截器实现分流：启动日志流、时间日志流
 *  将flume接收到的用户行为数据按照启动日志和事件日志进行分类，发送到不同的目的地
 *  注意：项目中先执行ETL拦截器，在执行分流拦截器
 */
public class LogTypeInterceptor implements Interceptor {
    //初始化方法：只执行一次
    @Override
    public void initialize() { }

    /**
     * @Param events:
     * @return java.util.List<org.apache.flume.Event>
     * @description TODO 系统调用方法：flume将采集到的数据封装到一个批次中，然后调用拦截器，校验完返回一个批次：
     */
    @Override
    public List<Event> intercept(List<Event> events) {
        ArrayList<Event> interceptors = new ArrayList<>();
        for (Event event : events) {
            Event resultEvent = intercept(event);
            interceptors.add(resultEvent);
        }
        return interceptors;
    }

    /**
     * @Param event:
     * @return org.apache.flume.Event
     * @description TODO 业务逻辑方法：
     *   对于flume来说，是基于事件机制的，会把每一条数据当做一个event事件来处理：
     *   Event对象包含: header是一个map，body是一个字节数组，
     *                body是真正传输的数据，header传输的数据不会被sink出去。
     *                header默认为null, 当需要实现分流时，可以设置header的值，作为标记：
     *   知识补充：https://blog.csdn.net/weixin_42946624/article/details/102622590
     */
    @Override
    public Event intercept(Event event) {
        //1、获取一条数据：
        byte[] body = event.getBody();
        String log = new String(body, Charset.forName("UTF-8"));
        //2、获取header：
        Map<String, String> headers = event.getHeaders();
        //3、设计分流标记：判断数据类型并向Header中赋值
        if (log.contains("start")) {
            headers.put("topic","topic_start");
        }else {
            headers.put("topic","topic_event");
        }
        return event;
    }

    //关闭资源：只执行一次
    @Override
    public void close() { }

    /**
     * 构建外部类对象
     */
    public static class Builder implements  Interceptor.Builder{
        // 拦截器对象实例化：
        @Override
        public Interceptor build() {
            return new LogTypeInterceptor();
        }
        // 拦截器参数配置：
        @Override
        public void configure(Context context) { }
    }
}
